package com.yss.netty;

import com.yss.exception.RemotingException;
import com.yss.exception.RemotingTimeOutException;
import com.yss.netty.codec.NettyDecode;
import com.yss.netty.codec.NettyEncode;
import com.yss.netty.config.NettyClientConfig;
import com.yss.netty.future.ResponseCallback;
import com.yss.netty.future.ResponseFuture;
import com.yss.netty.handler.NettyClientHandler;
import com.yss.netty.protocol.Message;
import com.yss.netty.util.Constants;
import com.yss.netty.util.Host;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class NettyRemotingClient {

    /**
     * Worker group
     */
    private final EventLoopGroup workerGroup;

    /**
     * Client config
     */
    private final NettyClientConfig clientConfig;

    /**
     * Boot strap
     */
    private final Bootstrap bootstrap;

    /**
     * Start flag
     */
    private final AtomicBoolean startFlag;

    /**
     * ScheduledPool for clear timeout future
     */
    private final ScheduledExecutorService responseFutureExecutor;

    private ConcurrentHashMap<Host, Channel> channels;

    /**
     * Client handler
     */
    private NettyClientHandler nettyClientHandler;

    public NettyRemotingClient(final NettyClientConfig clientConfig) {
        this.bootstrap = new Bootstrap();
        this.clientConfig = clientConfig;
        this.channels = new ConcurrentHashMap<>();
        this.nettyClientHandler = new NettyClientHandler();
        this.startFlag = new AtomicBoolean(false);
        this.responseFutureExecutor = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "ResponseFutureExecutor"));

        this.workerGroup = new NioEventLoopGroup(this.clientConfig.getWorkerThreadNum(), new ThreadFactory() {
            private final AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyClient_%d", this.threadIndex.incrementAndGet()));
            }
        });
    }

    public void start() {
        this.bootstrap
                .group(this.workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, this.clientConfig.SO_KEEPALIVE)
                .option(ChannelOption.TCP_NODELAY, this.clientConfig.isTcpNoDelay())
                .option(ChannelOption.SO_SNDBUF, this.clientConfig.getSendBufferSize())
                .option(ChannelOption.SO_RCVBUF, this.clientConfig.getReceiveBufferSize())
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.clientConfig.getConnectTimeoutMillis())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        initPipelineHandle(ch);
                    }
                });
        this.startFlag.compareAndSet(false, true);
        // clear future
        this.responseFutureExecutor.scheduleAtFixedRate(ResponseFuture::scanTimeOutFutures, 10, 10, TimeUnit.SECONDS);
    }

    private void initPipelineHandle(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("decoder", new NettyDecode())
                .addLast("encoder", new NettyEncode())
                .addLast("client-idle-handler", new IdleStateHandler(Constants.NETTY_CLIENT_HEART_BEAT_TIME, 0, 0, TimeUnit.MILLISECONDS))
                .addLast("netty-client-handler", nettyClientHandler);
    }

    /**
     * Just send message to other server,
     * and do not wait response message.
     *
     * @param host
     * @param message
     */
    public void send(Host host, Message message) throws RemotingException {
        Channel channel = getChannel(host);
        if (channel == null) {
            throw new RemotingException(String.format("connect to : %s fail", host));
        } else {
            try {
                ChannelFuture future = channel.writeAndFlush(message);
                future.sync();
                if (!future.isSuccess()) {
                    String errMsg = String.format("\nSend message to address %s failed", future.cause().toString());
                    log.error(errMsg);
                    throw new RemotingException(errMsg);
                }
            } catch (Exception e) {
                throw new RemotingException(e);
            }
        }
    }

    /**
     * Send message sync.
     * the method will blocking until get response from other server.
     *
     * @param host        target
     * @param message     message
     * @param waitTimeOut read response time out
     * @throws RemotingException        Send Message failed
     * @throws RemotingTimeOutException Read response timeout
     */
    public Message sendSync(final Host host, final Message message, final long waitTimeOut) throws RemotingException, RemotingTimeOutException {
        Channel channel = getChannel(host);
        if (channel == null) {
            throw new RemotingException(String.format("connect to : %s fail", host));
        } else {
            ResponseFuture responseFuture = new ResponseFuture(waitTimeOut);
            responseFuture.addFuture(message.getOpaque());

            channel.writeAndFlush(message).addListener(future -> {
                if (future.isSuccess()) {
                    responseFuture.setOk(true);
                    return;
                } else {
                    responseFuture.setOk(false);
                    responseFuture.setErrCause(future.cause().getMessage());
                    responseFuture.putResponse(null);
                }
            });
            Message response = responseFuture.waitResponse();
            // This requires the server to return a message, even if it is an empty string
            if (response == null) {
                if (responseFuture.isOk()) {
                    throw new RemotingTimeOutException(String.format("Message read time out from server : %s", host.getAddress()));
                }
                throw new RemotingException(String.format("Send message to server : %s failed. The cause is : %s", host.getAddress(), responseFuture.getErrCause()));
            }
            return response;
        }
    }

    /**
     * Send message async
     *
     * @param host
     * @param message          msg
     * @param responseCallback callback can be null
     * @throws RemotingException
     */
    public void sendAsync(final Host host, final Message message, final long waitTimeOut, final ResponseCallback responseCallback) throws RemotingException {
        Channel channel = getChannel(host);
        if (channel == null) {
            throw new RemotingException(String.format("connect to : %s fail", host));
        } else {
            channel.writeAndFlush(message).addListener(future -> {
                ResponseFuture responseFuture = new ResponseFuture(waitTimeOut);
                responseFuture.addFuture(message.getOpaque());
                responseFuture.setCallback(responseCallback);
                if (!future.isSuccess()) {
                    responseFuture.putResponse(null);
                }
            });
        }
    }

    private Channel getChannel(Host host) {
        Channel channel = channels.get(host);
        if (channel != null && channel.isActive()) {
            return channel;
        } else {
            // must clear death Channel
            channels.remove(host);
        }
        return createChannel(host);
    }

    private Channel createChannel(Host host) {
        ChannelFuture future;
        synchronized (this.bootstrap) {
            Channel findChannelAgain = channels.get(host);
            if (findChannelAgain == null) {
                try {
                    future = this.bootstrap.connect(new InetSocketAddress(host.getIp(), host.getPort()));
                    future.sync();
                    if (future.isSuccess()) {
                        Channel channel = future.channel();
                        channels.put(host, channel);
                        return channel;
                    }
                } catch (InterruptedException e) {
                    log.error("\nClient connect to address : {} failed", host.getAddress());
                }
            } else {
                return findChannelAgain;
            }
        }
        return null;
    }

    public void close() {
        if (this.startFlag.compareAndSet(true, false)) {
            if (this.workerGroup != null) {
                this.workerGroup.shutdownGracefully();
            }
            log.info("\nNetty client closed");
        }
    }

}
